package org.example.kafka_test.websocket;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.*;

import static org.example.kafka_test.webConf.WebSocketServer.sendMessage;


public class SocketConsumer extends Thread {
@Override
public void run(){
    // 创建Properties对象，用于配置Kafka消费者的属性
    Properties prop = new Properties();
    System.out.println("启动kafka消费者....");

    // 配置Kafka消费者的属性
    // 指定Kafka服务器的地址和端口
    prop.put("bootstrap.servers","47.108.77.113:19092");
    // 指定消费者组ID
    prop.put("group.id","socket");
    // 配置键的反序列化方式
    prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    // 配置值的反序列化方式
    prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    // 创建Kafka消费者对象
    Consumer consumer = new KafkaConsumer(prop);
    // 订阅主题为"zeek_test"的消息
    consumer.subscribe(Arrays.asList("zeek_test"));

    // 无限循环，持续消费Kafka主题中的消息
    while (true) {
        // 拉取消息，最大时长等待时间为100毫秒
        ConsumerRecords<String, String> c = consumer.poll(100);
        // 遍历拉取到的消息
        for(ConsumerRecord<String, String> c1: c) {
            // 打印消息内容
            System.out.println(c1.value());
            // 将消息内容发送到名为"socket"的目标
            sendMessage("socket",c1.value());
        }
    }
}

}